package com.xiao.dynamic_rule.service;

import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.RuleParam;
import com.xiao.dynamic_rule.pojo.RuleParamAtomic;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.state.ListState;

import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

/**
 * @description: 分路查询路由
 * @author: 啸
 * @date: 2022/8/23 10:11
 */
public class QueryRouterV3 {

    UserProfileQueryServiceHbaseImpl userProfileQueryServiceHbase;
    UserActionCountQueryServiceStateImpl userActionCountQueryServiceState;
    UserActionCountQueryServiceClickHouseImpl userActionCountQueryServiceClickHouse;
    UserActionSequenceQueryServiceStateImpl userActionSequenceQueryServiceState;
    UserActionSequenceQueryServiceClickhouseImpl userActionSequenceQueryServiceClickhouse;


    // 在构造方法中初始化连接
    public QueryRouterV3() throws IOException, SQLException, ClassNotFoundException {
        userProfileQueryServiceHbase = new UserProfileQueryServiceHbaseImpl();
        userActionCountQueryServiceState = new UserActionCountQueryServiceStateImpl();
        userActionCountQueryServiceClickHouse = new UserActionCountQueryServiceClickHouseImpl();
        userActionSequenceQueryServiceState = new UserActionSequenceQueryServiceStateImpl();
        userActionSequenceQueryServiceClickhouse = new UserActionSequenceQueryServiceClickhouseImpl();
    }

    /**
     * 画像查询路由
     *
     * @param ruleParam 规则条件
     * @param logBean   flink stream
     * @return 是否满足
     */
    public boolean userProfileQuery(RuleParam ruleParam, LogBean logBean) {
        //画像条件
        boolean profileMatch = userProfileQueryServiceHbase.judgeQueryProfileCondition(logBean.getDeviceId(), ruleParam);
        return profileMatch;
    }

    /**
     * 行为次数查询路由
     * TODO 将规则条件分开查询，最后没有累加？？？ 查看的在实现类中进行了累加（还要细捋一捋）
     */
    public boolean countConditionQuery(RuleParam ruleParam, LogBean logBean, ListState<LogBean> eventState) throws Exception {

        // 获取分界点（当前时间向下取整，再向前减1小时）
        long splitPoint = DateUtils.addHours(DateUtils.ceiling(new Date(), Calendar.HOUR), -2).getTime();

        // 为了减小clickhouse的压力，将条件根据分界点分为三段，
        // startTime < 分界点就去clickhouse中查，startTime > 分界点就在state中查
        // 时间跨度包含分界点，则先去state查，如果不满足则在去clickhouse中查，然后state结果和clickhouse结果整合来判断是否满足条件
        ArrayList<RuleParamAtomic> nearRangeParams = new ArrayList<>();
        ArrayList<RuleParamAtomic> farRangeParams = new ArrayList<>();
        ArrayList<RuleParamAtomic> crossRangeParams = new ArrayList<>();

        // 拿到条件中的次数条件
        List<RuleParamAtomic> userActionCountParams = ruleParam.getUserActionCountParams();

        // 根据时间来分为三组
        for (RuleParamAtomic userActionCountParam : userActionCountParams) {
            if (userActionCountParam.getRangeEnd() < splitPoint) {
                // 放到nearRangeParams中
                nearRangeParams.add(userActionCountParam);
            } else if (userActionCountParam.getRangeStart() >= splitPoint) {
                // 放到farRangeParams中
                farRangeParams.add(userActionCountParam);
            } else {
                // 放到crossRangParams中
                crossRangeParams.add(userActionCountParam);
            }
        }

        /**
         * 近期查询
         */
        if (nearRangeParams.size() > 0) {
            //将规则总参数对象中的用户次数类规则覆盖为 近期条件组
            ruleParam.setUserActionCountParams(nearRangeParams);
            // 在state中查询
            boolean nearCountMatch = userActionCountQueryServiceState.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            return nearCountMatch;
        }

        /**
         * 远期查询
         */
        if (farRangeParams.size() > 0) {
            //覆盖参数
            ruleParam.setUserActionCountParams(farRangeParams);
            //在clickhouse中查询
            boolean farCountMatch = userActionCountQueryServiceClickHouse.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            return farCountMatch;
        }

        /**
         * 跨界查询
         */
        for (RuleParamAtomic crossRangeParam : crossRangeParams) {
            //首先将原来规则的 startTIme保存下来
            long originRangeStart = crossRangeParam.getRangeStart();

            //将规则中的startTime设置成分界点，查state
            crossRangeParam.setRangeStart(splitPoint);
            boolean crossCountStateMatch = userActionCountQueryServiceState.queryActionCount(logBean.getDeviceId(), eventState, crossRangeParam);
            if (crossCountStateMatch) continue;

            //如果上面不满足，则将startEnd换为分界点，去clickhouse中查询
            crossRangeParam.setRangeStart(originRangeStart);
            crossRangeParam.setRangeEnd(splitPoint);
            boolean crossCountClickhouseMatch = userActionCountQueryServiceClickHouse.queryActionCount(logBean.getDeviceId(), eventState, crossRangeParam);
            return crossCountClickhouseMatch;

        }

        return true;
    }


    /**
     * 行为序列查询路由
     */
    public boolean sequenceConditionQuery(RuleParam ruleParam, LogBean logBean, ListState<LogBean> eventState) throws Exception {

        //取出规则中的行为序列条件
        List<RuleParamAtomic> userActionSequenceParams = ruleParam.getUserActionSequenceParams();

        //取出本身条件的开始和结束时间
        long rangeStart = userActionSequenceParams.get(0).getRangeStart();
        long rangeEnd = userActionSequenceParams.get(0).getRangeEnd();

        //序列条件中有条件才计算
        if (userActionSequenceParams != null && userActionSequenceParams.size() > 0) {
            //取到分界点时间（和次数类计算应该有时间差，所以单独创建）
            long splitPoint = DateUtils.addHours(DateUtils.ceiling(new Date(), Calendar.HOUR), -2).getTime();

            //取出序列规则中的总步骤数
            int totalSize = ruleParam.getUserActionSequenceParams().size();

            if (userActionSequenceParams.get(0).getRangeStart() > splitPoint) {
                //全部在state中查
                return userActionSequenceQueryServiceState.queryActionSequence(logBean.getDeviceId(), eventState, ruleParam);
            } else if (userActionSequenceParams.get(0).getRangeEnd() < splitPoint) {
                //全部在clickhouse中查
                return userActionCountQueryServiceClickHouse.queryActionCounts(logBean.getDeviceId(), eventState, ruleParam);
            } else {
                //state中一部分，clickhouse中一部分

                //将条件开始时间换为分界点
                modifyTimeRange(userActionSequenceParams, splitPoint, userActionSequenceParams.get(0).getRangeEnd());
                //先查state，如果state中已经满足了条件，则就不必去clickhouse中查了
                boolean sequenceStateMatch = userActionSequenceQueryServiceState.queryActionSequence("", eventState, ruleParam);
                if (sequenceStateMatch) return true;

                //将条件结束时间换为分界点
                modifyTimeRange(userActionSequenceParams,userActionSequenceParams.get(0).getRangeStart(),splitPoint);
                //如果在state中没有查到，则去clickhouse中查，clickhouse满足也行
                boolean sequenceClickhouseMatch = userActionSequenceQueryServiceClickhouse.queryActionSequence(logBean.getDeviceId(), null, ruleParam);
                //将最大步骤数先拿出来，后面整合
                int clickhouseMaxStep = ruleParam.getUserActionSequenceQueriedMaxStep();
                if (sequenceClickhouseMatch) return true;

                //如果clickhouse也不能全部满足，则将条件中clickhouse没查到的条件去state查，然后组合
                //将条件的开始时间换为分界点
                modifyTimeRange(userActionSequenceParams, splitPoint, rangeEnd);

                //现在将clickhouse已经查到满足的条件去除，只在state中查询还未满足的条件
                ruleParam.setUserActionSequenceParams(userActionSequenceParams.subList(ruleParam.getUserActionSequenceQueriedMaxStep(), userActionSequenceParams.size()));

                //查询截断后的条件
                boolean b = userActionSequenceQueryServiceState.queryActionSequence("", eventState, ruleParam);

                //将结果整合，放回规则对象中
                ruleParam.setUserActionSequenceQueriedMaxStep(ruleParam.getUserActionSequenceQueriedMaxStep() + clickhouseMaxStep);
                return ruleParam.getUserActionSequenceQueriedMaxStep() >= totalSize;
            }
        }
        return false;
    }

    /**
     * 更新条件的时间窗口的工具方法
     *
     * @param userActionSequenceParams 行为序列条件
     * @param newStart                 条件中新的开始时间
     * @param newEnd                   条件中新的结束时间
     */
    public void modifyTimeRange(List<RuleParamAtomic> userActionSequenceParams, long newStart, long newEnd) {
        for (RuleParamAtomic userActionSequenceParam : userActionSequenceParams) {
            userActionSequenceParam.setRangeStart(newStart);
            userActionSequenceParam.setRangeEnd(newEnd);
        }
    }
}
